package club.vann.kafka.mock.service.strategy;

import club.vann.kafka.mock.common.model.info.DynamicTableInfo;
import club.vann.kafka.mock.common.model.info.FrontEndTable;
import club.vann.kafka.mock.common.model.request.StrategyParentActionRequest;
import club.vann.kafka.mock.kafka.KafkaProducer;
import club.vann.kafka.mock.kafka.KafkaTopicFactory;
import com.alibaba.fastjson.JSON;
import com.cyberx.lunyu.StrategyParentActionOuterClass;
import com.cyberx.lunyu.StrategySummaryOuterClass;
import com.cyberx.lunyu.TableDataOuterClass;
import com.googlecode.protobuf.format.JsonFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;

@RequiredArgsConstructor
@Service
public class StrategyService {
  private final KafkaProducer kafkaProducer;

  /**
   * 模拟推送策略汇总信息。
   *
   * @param strategySummaryInfoString 汇总信息字符串
   */
  public void strategySummaryInfo(String strategySummaryInfoString) {
    StrategySummaryOuterClass.StrategySummary.Builder strategySummary =
        StrategySummaryOuterClass.StrategySummary.newBuilder();
    try {
      JsonFormat.merge(strategySummaryInfoString, strategySummary);
      this.kafkaProducer.send(
          KafkaTopicFactory.STRATEGY_SUMMARY_OUTPUT,
          strategySummary.getStrategySpec(),
          strategySummary.build().toByteArray());
    } catch (JsonFormat.ParseException e) {
      throw new RuntimeException(e);
    }
  }

  /** 模拟推送策略指标信息。 */
  public void strategyIndicator(String tableDataString) {
    DynamicTableInfo dynamicTableInfo = JSON.parseObject(tableDataString, DynamicTableInfo.class);
    String tableName = dynamicTableInfo.getTableName();
    List<String> tableDataList = dynamicTableInfo.getTableData();
    String headerList = tableDataList.getFirst();
    List<String> rowDataList = tableDataList.subList(1, tableDataList.size());
    TableDataOuterClass.TableData tableData = Util.ofTableData(tableName, headerList, rowDataList);
    this.kafkaProducer.send(
        KafkaTopicFactory.TABLE_DATA_OUTPUT + "." + tableData.getTableName(),
        tableData.getTableName(),
        tableData.toByteArray());
  }

  /**
   * 模拟推送Parent Action信息
   *
   * @param request 请求信息
   */
  public void parentAction(StrategyParentActionRequest request) {
    StrategyParentActionOuterClass.StrategyParentAction strategyParentAction =
        StrategyParentActionOuterClass.StrategyParentAction.newBuilder()
            .setAction(request.getAction())
            .setRequestId(UUID.randomUUID().toString())
            .setStrategySpec(request.getStrategySpec())
            .setSubStrategySpec(request.getSubStrategySpec())
            .putAllConfig(request.getConfigMap())
            .setSendTime(System.currentTimeMillis())
            .setSubConfigType("TWAP")
            .build();
    this.kafkaProducer.send(
        KafkaTopicFactory.STRATEGY_PARENT_ACTION_OUTPUT,
        request.getStrategySpec(),
        strategyParentAction.toByteArray());
  }

  private static class Util {
    public static TableDataOuterClass.TableData ofTableData(
        String tableName, String headerList, List<String> rowList) {
      FrontEndTable.DataType[] dataTypeArray = ofDataTypeArray(headerList);
      return TableDataOuterClass.TableData.newBuilder()
          .setTableName(tableName)
          .setTimestamp(System.currentTimeMillis())
          .addAllColumn(ofColumnList(headerList))
          .addAllRow(ofRowList(dataTypeArray, rowList))
          .build();
    }

    public static List<TableDataOuterClass.Column> ofColumnList(String headerList) {
      String[] headerArray = headerList.split(",");
      List<TableDataOuterClass.Column> columnList = new ArrayList<>();
      for (String header : headerArray) {
        TableDataOuterClass.Column column = ofColumn(header);
        columnList.add(column);
      }
      return columnList;
    }

    public static TableDataOuterClass.Column ofColumn(String columnString) {
      String[] split = columnString.split("\\|");
      return TableDataOuterClass.Column.newBuilder()
          .setColumnName(split[0])
          .setDataType(split[1])
          .build();
    }

    public static List<TableDataOuterClass.Row> ofRowList(
        FrontEndTable.DataType[] dataTypeArray, List<String> rowList) {
      List<TableDataOuterClass.Row> list = new ArrayList<>();
      for (String rowString : rowList) {
        TableDataOuterClass.Row row = ofRow(dataTypeArray, rowString);
        list.add(row);
      }
      return list;
    }

    public static TableDataOuterClass.Row ofRow(
        FrontEndTable.DataType[] headerDataTypeArray, String rowString) {
      String[] columnArray = rowString.split(",");
      List<TableDataOuterClass.Cell> list = new ArrayList<>();
      for (int index = 0; index < columnArray.length; index++) {
        String column = columnArray[index];
        FrontEndTable.DataType dataType = headerDataTypeArray[index];
        TableDataOuterClass.Cell cell = ofCell(dataType, column);
        list.add(cell);
      }
      return TableDataOuterClass.Row.newBuilder().addAllCell(list).build();
    }

    public static TableDataOuterClass.Cell ofCell(
        FrontEndTable.DataType dataType, String cellString) {
      switch (dataType) {
        case DOUBLE:
          return TableDataOuterClass.Cell.newBuilder()
              .setDoubleData(Double.parseDouble(cellString))
              .build();
        case STRING:
          return TableDataOuterClass.Cell.newBuilder().setStringData(cellString).build();
        case TIMESTAMP:
          return TableDataOuterClass.Cell.newBuilder()
              .setLongData(Long.parseLong(cellString))
              .build();
        default:
          return TableDataOuterClass.Cell.newBuilder().setStringData(cellString).build();
      }
    }

    public static FrontEndTable.DataType[] ofDataTypeArray(String headerList) {
      String[] headerArray = headerList.split(",");
      FrontEndTable.DataType[] dataTypeArray = new FrontEndTable.DataType[headerArray.length];
      for (int index = 0; index < headerArray.length; index++) {
        String header = headerArray[index];
        String[] split = header.split("\\|");
        FrontEndTable.DataType dataType = FrontEndTable.DataType.valueOf(split[1]);
        dataTypeArray[index] = dataType;
      }
      return dataTypeArray;
    }
  }
}
